package com.cx.base.day25_Rocketmq.producer.filterProducer;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * @Author: songyahui
 * @Date: 2020/9/27 19:31
 * @Version 1.0
 * <p>
 * <p>
 * 创建消息生产者，并且制定生产者组名
 * 指定namesrv地址
 * 启动producer
 * 创建消息对象，指定主题topic,tag和消息体
 * 发送消息
 * 关闭producer
 */
public class FilterProducer {
    public static void main(String[] args) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
//        1.创建消息生产者，并且制定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("group1");
//        设置超时时间
        producer.setSendMsgTimeout(8000);
//        2.指定namesrv地址
        producer.setNamesrvAddr("192.168.88.133:9876;192.168.88.134:9876");
//        producer.setNamesrvAddr("192.168.48.128:9876;192.168.48.137:9876");
//        producer.setNamesrvAddr("47.96.165.93:9876;106.52.131.197:9876");
//        3.启动producer
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //测试消费端的orderly监听模式是否会一直重试，为方便测试这里设置i<1
        for (int i = 0; i < 1; i++) {
//        4.创建消息对象，指定主题topic,tag和消息体
//            Message msg = new Message("filterTopic","tag1",("hello world"+i).getBytes());
            Message msg = new Message("filterTopic", ("hello world" + i).getBytes());
            msg.putUserProperty("i", String.valueOf(i));
//        5.发送消息
            SendResult result = producer.send(msg);
            SendStatus status = result.getSendStatus();
            String msgId = result.getMsgId();
            int queueId = result.getMessageQueue().getQueueId();
            System.out.println("send status:" + result + ",message ID:" + msgId + ",queue:" + queueId);
            TimeUnit.SECONDS.sleep(1);
            System.out.println(i);
        }
//                * 关闭producer
        producer.shutdown();
    }
}

